/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package DAO;

import AutoIncrementClient.AutoIncrementEnvoker;
import Common.Constants;
import Common.GeneralHandling;
import POJO.CounterPOJO;
import POJO.StatusPOJO;
import POJO.TagNameAndCounter;
import SearchSolr.solrSearch;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.*;
import org.apache.thrift.TException;

/**
 *
 * @author node1
 */
public class StatusDAO {

    public static ByteBuffer colParent = ByteBuffer.wrap("Status".getBytes());
    public static ByteBuffer colNameContent = ByteBuffer.wrap("content".getBytes());
    public static ByteBuffer colNameCreateDate = ByteBuffer.wrap("createDate".getBytes());
    public static ByteBuffer colNameModifyDate = ByteBuffer.wrap("modifyDate".getBytes());
    public static ByteBuffer colNameLikeCount = ByteBuffer.wrap("likeCount".getBytes());
    public static ByteBuffer colNameDislikeCount = ByteBuffer.wrap("dislikeCount".getBytes());
    public static ByteBuffer colNameIsPublic = ByteBuffer.wrap("state".getBytes());
    public static ByteBuffer colNameTags = ByteBuffer.wrap("tags".getBytes());
    public static String colParentS = "Status";
    public static String colNameContentS = "content";
    public static String colNameCreateDateS = "createDate";
    public static String colNameModifyDateS = "modifyDate";
    public static String colNameLikeCountS = "likeCount";
    public static String colNameDislikeCountS = "dislikeCount";
    public static String colNameIsPublicS = "state";
    public static String colNameTagsS = "tags";
    private static Cassandra.Client client;
    private static StatusDAO instance;

    public static StatusDAO getInstance(Cassandra.Client _client) {
        if (instance != null) {
            return instance;
        }
        instance = new StatusDAO();
        client = _client;
        return instance;
    }

    public boolean insertStatus(StatusPOJO status) throws Exception {
        boolean result = false;
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            //Lay id autoincrement
            String statusID = GeneralHandling.toStringFromLong(AutoIncrementEnvoker.getApi().getAutoIncrement("Status"));
            ByteBuffer statusIDByteBuffer = GeneralHandling.toByteBuffer(statusID);
            //Add các column vào CF.
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameContent, status.content), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameTags, status.tags), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameLikeCount, Integer.toString(status.likeCount)), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameDislikeCount, Integer.toString(status.dislikeCount)), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameCreateDate, ""), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameModifyDate, ""), Constants.CL);
            client.insert(statusIDByteBuffer, parent, GeneralHandling.createCol(colNameIsPublic, (status.isPublic == true) ? "public" : "private"), Constants.CL);

            //Cập nhật list_status_ID cho tag tương ứng.
            String[] tags = status.getTags().split(",");
            TagDAO tag_DAO = TagDAO.getInstance(client);
            for (int i = 0; i < tags.length; i++) {
                if (!tag_DAO.insertStatusIntoTag(statusID, tags[i])) {
                    return result;
                }
            }
            CounterDAO.Inscrease(CounterDAO.colNameStatus, client);

            //Cap nhat solr.
            solrSearch solr = new solrSearch();
            String[] temp = status.tags.split(",");
            ArrayList<String> listtag = new ArrayList<String>();
            listtag.addAll(Arrays.asList(temp));
            solr.insertStatus(statusID, status.content, status.createDate, status.modifyDate, status.isPublic == true ? "1" : "0", listtag);

            result = true;
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }

        return result;
    }

    public boolean updateContentStatus(String statusID, String content) throws Exception {
        boolean result = false;
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            ByteBuffer status_ID = GeneralHandling.toByteBuffer(statusID);

            //Add các column vào CF.
            client.insert(status_ID, parent, GeneralHandling.createCol(colNameContent, content), Constants.CL);

            //Update Solr
            solrSearch solr = new solrSearch();
            String[] status = solr.SearchStatusById(statusID);
            String[] temp = status[5].split(",");
            ArrayList<String> listtag = new ArrayList<String>();
            listtag.addAll(Arrays.asList(temp));
            solr.insertStatus(statusID, content, status[2], status[3], status[4], listtag);

            result = true;

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }

        return result;
    }

    public boolean updateTagsOnAStatus(String statusID, String tags) throws Exception {
        try {
            //Xoa statusID trong cac tag truoc
            StatusPOJO statusPOJO = selectStatusWithID(statusID);
            String listTagOld = statusPOJO.tags;

            String[] temp = listTagOld.split(",");
            ArrayList<String> listtag = new ArrayList<String>();
            TagDAO tag = TagDAO.getInstance(client);

            for (int i = 0; i < temp.length; i++) {
                tag.removeStatusInATag(temp[i], statusID);
            }


            //Them tags moi vao status
            ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
            ColumnParent parent = new ColumnParent(colParentS);
            client.insert(rowKey, parent, GeneralHandling.createCol(colNameTags, tags), Constants.CL);


            //Update Solr, update status trong tags
            solrSearch solr = new solrSearch();
            String[] status = solr.SearchTagById(tags);
            temp = tags.split(",");
            listtag = new ArrayList<String>();
            for (int i = 0; i < temp.length; i++) {
                listtag.add(temp[i]);
                tag.insertStatusIntoTag(statusID, temp[i]);
            }
            if (status != null) {
                solr.insertStatus(statusID, status[1], status[2], status[3], status[4], listtag);
            }
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return true;
    }

    public int updateTagsOnListStatus(List<String> listStatusID, String tags) throws Exception {
        for (String statusID : listStatusID) {
            try {
                ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
                ColumnParent parent = new ColumnParent(colParentS);
                client.insert(rowKey, parent, GeneralHandling.createCol(colNameTags, tags), Constants.CL);

                //Update Solr
                solrSearch solr = new solrSearch();
                String[] status = solr.SearchTagById(tags);
                String[] temp = tags.split(",");
                ArrayList<String> listtag = new ArrayList<String>();
                TagDAO tag = TagDAO.getInstance(client);
                for (int i = 0; i < temp.length; i++) {
                    listtag.add(temp[i]);
                    tag.insertStatusIntoTag(statusID, temp[i]);
                }
                solr.insertStatus(statusID, status[1], status[2], status[3], status[4], listtag);

            } catch (InvalidRequestException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return 1;
    }

    public void createCache(List<String> listMem) throws UnavailableException {
        TagDAO tag_DAO = TagDAO.getInstance(client);
        List<TagNameAndCounter> listtag = tag_DAO.selectAllTagNameAndCounter();
        for (TagNameAndCounter tagNameAndCounter : listtag) {
            if (!cachedStatusFromTag(tagNameAndCounter.tagName, listMem)) {
                System.out.println("Không thể tạo cached");
            }
        }
        if (!selectTopStatus_cache(20, listMem)) {
            System.out.println("Không thể tạo cached");
        }
        if (!selectAllStatus_cache(listMem)) {
            System.out.println("Không thể tạo cached");
        }
    }

    public void updateListTag(List<ByteBuffer> liststatusID, String nameTag) {
        for (ByteBuffer rowKey : liststatusID) {
            try {
                SlicePredicate predicate = new SlicePredicate();
                predicate.addToColumn_names(colNameTags);
                ColumnParent parent = new ColumnParent(colParentS);

                List<ColumnOrSuperColumn> result = client.get_slice(rowKey, parent, predicate, Constants.CL);
                String tags = GeneralHandling.toString(result.get(0).column.value);
                tags = tags.replaceAll("," + nameTag, "");
                tags = tags.replaceAll(nameTag + ",", "");
                tags = tags.replaceAll(nameTag, "");
                client.insert(rowKey, parent, GeneralHandling.createCol(colNameTags, tags), Constants.CL);

            } catch (InvalidRequestException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    public void removeTagInListTagOnListStatus(List<ByteBuffer> listStatusID, String tagName) {
        for (ByteBuffer rowKey : listStatusID) {
            try {
                SlicePredicate predicate = new SlicePredicate();
                predicate.addToColumn_names(colNameTags);
                ColumnParent parent = new ColumnParent(colParentS);

                List<ColumnOrSuperColumn> result = client.get_slice(rowKey, parent, predicate, Constants.CL);
                String tags = GeneralHandling.toString(result.get(0).column.value);
                tags = tags.replaceAll("," + tagName, "");
                tags = tags.replaceAll(tagName + ",", "");
                tags = tags.replaceAll(tagName, "");
                client.insert(rowKey, parent, GeneralHandling.createCol(colNameTags, tags), Constants.CL);

            } catch (InvalidRequestException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    public void removeTagInListTagOnAStatus(String statusID, String tagName) {
        try {
            ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameTags);
            ColumnParent parent = new ColumnParent(colParentS);

            List<ColumnOrSuperColumn> result = client.get_slice(rowKey, parent, predicate, Constants.CL);
            if (!result.isEmpty()) {
                String tags = GeneralHandling.toString(result.get(0).column.value);
                tags = tags.replaceAll("," + tagName, "");
                tags = tags.replaceAll(tagName + ",", "");
                tags = tags.replaceAll(tagName, "");
                client.insert(rowKey, parent, GeneralHandling.createCol(colNameTags, tags), Constants.CL);
            }

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public boolean updateIsPublicOnAStatus(String statusID, boolean state) throws Exception {
        boolean result = false;
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            ByteBuffer status_ID = GeneralHandling.toByteBuffer(statusID);

            //Add các column vào CF.
            client.insert(status_ID, parent, GeneralHandling.createCol(colNameIsPublic, (state == true) ? "public" : "private"), Constants.CL);

            //Update Solr
            solrSearch solr = new solrSearch();
            String[] status = solr.SearchStatusById(statusID);
            String[] temp = status[1].split(",");
            ArrayList<String> listtag = new ArrayList<String>();
            listtag.addAll(Arrays.asList(temp));
            solr.insertStatus(statusID, status[1], status[2], status[3], state == true ? "1" : "0", listtag);

            result = true;

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }

        return result;
    }

    public int updateIsPublicOnListStatus(List<String> listStatusID, boolean state) throws Exception {
        for (String statusID : listStatusID) {
            try {
                ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
                ColumnParent parent = new ColumnParent(colParentS);
                client.insert(rowKey, parent, GeneralHandling.createCol(colNameIsPublic, (state == true) ? "public" : "private"), Constants.CL);

                //Update Solr
                solrSearch solr = new solrSearch();
                String[] status = solr.SearchStatusById(statusID);
                String[] temp = status[1].split(",");
                ArrayList<String> listtag = new ArrayList<String>();
                listtag.addAll(Arrays.asList(temp));
                solr.insertStatus(statusID, status[1], status[2], status[3], state == true ? "1" : "0", listtag);

            } catch (InvalidRequestException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return 1;
    }

    public boolean likeStatus(String statusID) {
        boolean result = false;
        try {
            ByteBuffer id = Common.GeneralHandling.toByteBuffer(statusID);
            ColumnParent parent = new ColumnParent(colParentS);
            ColumnPath path = new ColumnPath(colParentS);
            path.column = colNameLikeCount;
            int likeCountCurrent = 0;
            ColumnOrSuperColumn status;
            try {
                status = client.get(id, path, Common.Constants.CL);
                likeCountCurrent = Integer.parseInt(Common.GeneralHandling.toString(status.column.value));

            } catch (NotFoundException ex) {
                likeCountCurrent = 0;
            }
            Column newcol = GeneralHandling.createCol(colNameLikeCount, Integer.toString(likeCountCurrent + 1));
            client.insert(id, parent, newcol, Common.Constants.CL);
            result = true;
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }

    public boolean dislikeStatus(String statusID) {
        boolean result = false;
        try {
            ByteBuffer id = Common.GeneralHandling.toByteBuffer(statusID);
            ColumnParent parent = new ColumnParent(colParentS);
            ColumnPath path = new ColumnPath(colParentS);
            path.column = colNameDislikeCount;
            int dislikeCountCurrent = 0;
            ColumnOrSuperColumn status;
            try {
                status = client.get(id, path, Common.Constants.CL);
                dislikeCountCurrent = Integer.parseInt(Common.GeneralHandling.toString(status.column.value));

            } catch (NotFoundException ex) {
                dislikeCountCurrent = 0;
            }
            Column newcol = GeneralHandling.createCol(colNameDislikeCount, Integer.toString(dislikeCountCurrent + 1));
            client.insert(id, parent, newcol, Common.Constants.CL);
            result = true;
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }

    public boolean deleteStatus(String statusID) {
        boolean result = false;
        try {
            ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
            ColumnPath path = new ColumnPath(colParentS);
            ColumnParent parent = new ColumnParent(colParentS);
            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameTags);
            long timestamp = System.currentTimeMillis();

            List<ColumnOrSuperColumn> col = client.get_slice(rowKey, parent, predicate, Constants.CL);
            if (!col.isEmpty()) {
                String listTag = GeneralHandling.toString(col.get(0).column.value);
                String[] tags = listTag.split(",");
                client.remove(rowKey, path, timestamp, ConsistencyLevel.ONE);

                TagDAO tag_DAO = TagDAO.getInstance(client);
                for (int i = 0; i < tags.length; i++) {
                    if (!"".equals(tags[i])) {
                        result = tag_DAO.removeStatusInATag(tags[i], statusID);
                    }
                }
                CounterDAO.Decrease(CounterDAO.colNameStatus, client);
                return result;
            }

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }

    public int deleteListStatus(List<String> listStatusId) {
        for (String statusID : listStatusId) {
            try {
                ByteBuffer rowKey = GeneralHandling.toByteBuffer(statusID);
                ColumnPath path = new ColumnPath(colParentS);
                ColumnParent parent = new ColumnParent(colParentS);
                SlicePredicate predicate = new SlicePredicate();
                predicate.addToColumn_names(colNameTags);
                long timestamp = System.currentTimeMillis();

                List<ColumnOrSuperColumn> col = client.get_slice(rowKey, parent, predicate, Constants.CL);
                String listTag = GeneralHandling.toString(col.get(0).column.value);
                String[] tags = listTag.split(",");
                client.remove(rowKey, path, timestamp, ConsistencyLevel.ONE);

                TagDAO tag_DAO = TagDAO.getInstance(client);
                for (int i = 0; i < tags.length; i++) {
                    tag_DAO.removeStatusInATag(tags[i], statusID);
                }
                CounterDAO.Decrease(CounterDAO.colNameStatus, client);

            } catch (InvalidRequestException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnavailableException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TimedOutException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (TException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return 1;
    }

    public StatusPOJO selectStatusWithID(String statusID) {
        StatusPOJO status = new StatusPOJO();
        try {
            ByteBuffer id = Common.GeneralHandling.toByteBuffer(statusID);
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = createSlicePredicateForStatus();

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(id, parent, predicate, Common.Constants.CL);
            if(listColOrSupCol.size() != 0){
                status = createStatusPOJOFromListColOrSupCol(statusID, listColOrSupCol);
            }else{
                status.statusID = "-1";
                status.content = "-1";
                status.isPublic = false;
            }
            Memcached.set(statusID, Constants.EXPIRE, status);

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return status;
    }

    public static StatusPOJO selectStatusWithID(String statusID, Cassandra.Client client) {
        StatusPOJO status = new StatusPOJO();
        try {
            ByteBuffer status_ID = GeneralHandling.toByteBuffer(statusID);
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = createSlicePredicateForStatus();

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(status_ID, parent, predicate, Common.Constants.CL);
            
            if(listColOrSupCol.size() != 0){
                status = createStatusPOJOFromListColOrSupCol(statusID, listColOrSupCol);
            }else{
                status.statusID = "-1";
                status.content = "-1";
                status.isPublic = false;
            }

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return status;
    }

    public List<StatusPOJO> selectStatusFromTag(String tagName) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        List<String> listStatusID = selectStatusIDFromTag(tagName);
        Memcached.set("StatusFromTag_" + Integer.toString(tagName.hashCode()), Constants.EXPIRE, listStatusID);
        listStatus = getListStatusFromListStatusID(listStatusID, client);
        return listStatus;
    }

    public boolean cachedStatusFromTag(String tagName, List<String> listMem) throws UnavailableException {
        List<String> listStatusID = selectStatusIDFromTag(tagName);
        Memcached.set("StatusFromTag_" + Integer.toString(tagName.hashCode()), Constants.EXPIRE, listStatusID);
        listMem.add("StatusFromTag_" + Integer.toString(tagName.hashCode()));
        return true;
    }
     public boolean cachedStatusFromTag(String tagName) throws UnavailableException {
        List<String> listStatusID = selectStatusIDFromTag(tagName);
        Memcached.set("StatusFromTag_" + Integer.toString(tagName.hashCode()), Constants.EXPIRE, listStatusID);
        return true;
    }

    public List<StatusPOJO> selectTopStatusFromTag(String tagName, int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            listStatus = selectStatusFromTag(tagName);
            listStatus = SortByLikeAndDislike(listStatus).subList(0, limit);
            List<String> listStatusID = new ArrayList<String>();
            for (StatusPOJO temp : listStatus) {
                listStatusID.add(temp.statusID);
            }
            Memcached.set("TopStatusFromTag_" + Integer.toString(tagName.hashCode()), Constants.EXPIRE, listStatusID);

        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public boolean selectTopStatusFromTag_cache(String tagName, int limit) {
        boolean result = false;
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            listStatus = selectStatusFromTag(tagName);
            if (listStatus.size() < limit) {
                limit = listStatus.size();
            }
            listStatus = SortByLikeAndDislike(listStatus).subList(0, limit);
            List<String> listStatusID = new ArrayList<String>();
            for (StatusPOJO temp : listStatus) {
                listStatusID.add(temp.statusID);
            }
            Memcached.set("TopStatusFromTag_" + Integer.toString(tagName.hashCode()), Constants.EXPIRE, listStatusID);

            result = true;
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }

    public List<StatusPOJO> selectTopStatus(int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            listStatus = selectAllStatus();
            listStatus = SortByLikeAndDislike(listStatus).subList(0, limit);
            List<String> listStatusID = new ArrayList<String>();
            for (StatusPOJO temp : listStatus) {
                listStatusID.add(temp.statusID);
            }
            Memcached.set("TopStatus", Constants.EXPIRE, listStatusID);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public boolean selectTopStatus_cache(int limit, List<String> listMem) {
        boolean result = false;
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            listStatus = selectAllStatus();
            listStatus = SortByLikeAndDislike(listStatus).subList(0, limit);
            List<String> listStatusID = new ArrayList<String>();
            for (StatusPOJO temp : listStatus) {
                listStatusID.add(temp.statusID);
            }
            Memcached.set("TopStatus", Constants.EXPIRE, listStatusID);
            listMem.add("TopStatus");
            result = true;
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }
     public boolean selectTopStatus_cache(int limit) {
        boolean result = false;
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            listStatus = selectAllStatus();
            listStatus = SortByLikeAndDislike(listStatus).subList(0, limit);
            List<String> listStatusID = new ArrayList<String>();
            for (StatusPOJO temp : listStatus) {
                listStatusID.add(temp.statusID);
            }
            Memcached.set("TopStatus", Constants.EXPIRE, listStatusID);
            result = true;
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }


    public List<StatusPOJO> selectRandomStatus(int limit) throws TException {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        int statusID = 0;
        int numberStatus = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status");
        List<String> listStatusID = new ArrayList<String>();
        Random ran = new Random();

        for (int i = 0; i < limit; i++) {
            statusID = ran.nextInt(numberStatus);
            listStatusID.add(Integer.toString(statusID));
        }
        listStatus = getListStatusFromListStatusID(listStatusID, client);
        return listStatus;
    }

    public List<StatusPOJO> selectRandomStatusV1(int limit) throws TException {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            int statusID = 0;
            int numberStatus = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status");

            List<ByteBuffer> listStatusID = new ArrayList<ByteBuffer>();
            Random ran = new Random();

            for (int i = 0; i < limit; i++) {
                statusID = ran.nextInt(numberStatus);
                listStatusID.add(Common.GeneralHandling.toByteBufferFromInt(statusID));
            }
            listStatus = getListStatusFromListStatusIDV1(listStatusID, client);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> selectRandomStatusV2(int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameContent);
            predicate.addToColumn_names(colNameTags);
            predicate.addToColumn_names(colNameIsPublic);

            int numberStatus = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status"); //Đã test ko có autoIncrement numberStatus=1211.
            String[] listID = new String[numberStatus];
            for (int i = 0; i < listID.length; i++) {
                listID[i] = Integer.toString(i);
            }
            for (int i = 0; i < listID.length - 1; i++) {
                for (int j = i + 1; j < listID.length; j++) {
                    if (listID[i].compareTo(listID[j]) == -1) {
                        String temp = listID[i];
                        listID[i] = listID[j];
                        listID[j] = temp;
                    }
                }
            }

            Random ran = new Random();
            String start_key = listID[listID.length - ran.nextInt(listID.length - limit)];
            String end_key = listID[0];

            KeyRange range = new KeyRange();
            range.start_key = Common.GeneralHandling.toByteBuffer(start_key);
            range.end_key = Common.GeneralHandling.toByteBuffer(end_key);
            range.count = limit;
            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            listStatus = getListStatusFromListKeySliceStatusForRanDom(listKeySlice);

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> selectRandomStatusV3(int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            int statusID = 0;

            List<ByteBuffer> listStatusID = new ArrayList<ByteBuffer>();
            Random ran = new Random();
            CounterPOJO counter = CounterDAO.getCounter(client);
            int countstatus = counter.getCounterStatus();

            int count = 0;
            int range = countstatus / limit;
            for (int i = 0; i < range; i++) {
                statusID = ran.nextInt(limit) + i * limit;
                StatusPOJO temp = selectStatusWithID(Integer.toString(statusID));
                if (!temp.isPublic) {
                    if (count >= (limit < range ? limit : range)) {
                        continue;
                    }
                    count++;
                    i--;
                    continue;
                }
                listStatusID.add(Common.GeneralHandling.toByteBufferFromInt(statusID));
                count = 0;
            }
            listStatus = getListStatusFromListStatusIDV1(listStatusID, client);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> selectRandomStatusFromTag(String tagName, int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ByteBuffer tagNameBB = Common.GeneralHandling.toByteBuffer(tagName);
            ColumnParent parent = new ColumnParent(TagDAO.colParentListStatusS);
            SlicePredicate predicate = new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.start = GeneralHandling.toByteBuffer("");
            sliceRange.finish = GeneralHandling.toByteBuffer("");
            predicate.slice_range = sliceRange;

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(tagNameBB, parent, predicate, Common.Constants.CL);

            List<ByteBuffer> listStatusID = new ArrayList<ByteBuffer>();

            for (int j = 0; j < listColOrSupCol.size(); j++) {
                ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                Column col = colOrSupCol.column;
                listStatusID.add(col.name);
            }
            //Xáo trôn list statusID
            Random random = new Random();
            int listSize = listStatusID.size() > limit ? limit : listStatusID.size();

            List<ByteBuffer> randomList = new ArrayList<ByteBuffer>();
            for (int i = 0; i < listSize; i++) {
                int pos = random.nextInt(listStatusID.size());
                randomList.add(listStatusID.get(pos));
                listStatusID.remove(pos);
            }
            listStatus = getListStatusFromListStatusIDV1(randomList, client);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> selectAllStatus() {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            SlicePredicate predicate = createSlicePredicateForStatus();

            KeyRange range = new KeyRange();
            range.start_key = Common.GeneralHandling.toByteBuffer("");
            range.end_key = Common.GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;
            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            listStatus = getListStatusFromListKeySliceStatus(listKeySlice);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public boolean selectAllStatus_cache(List<String> listMem) {
        boolean result = false;
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            SlicePredicate predicate = createSlicePredicateForStatus();

            KeyRange range = new KeyRange();
            range.start_key = Common.GeneralHandling.toByteBuffer("");
            range.end_key = Common.GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;
            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            listStatus = getListStatusFromListKeySliceStatus(listKeySlice);
            //Cache
            List<String> statusIds = new ArrayList<String>();
            for (int i = 0; i < listStatus.size(); i++) {
                Memcached.set(listStatus.get(i).statusID, Constants.EXPIRE, listStatus.get(i));
                statusIds.add(listStatus.get(i).statusID);
                listMem.add(listStatus.get(i).statusID);
            }
            //Set list status ID
            Memcached.set("StatusIds", Constants.EXPIRE, statusIds);
            listMem.add("StatusIds");
            result = true;
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }
    
    public boolean selectAllStatus_cache() {
        boolean result = false;
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            SlicePredicate predicate = createSlicePredicateForStatus();

            KeyRange range = new KeyRange();
            range.start_key = Common.GeneralHandling.toByteBuffer("");
            range.end_key = Common.GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;
            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            listStatus = getListStatusFromListKeySliceStatus(listKeySlice);
            //Cache
            List<String> statusIds = new ArrayList<String>();
            for (int i = 0; i < listStatus.size(); i++) {
                Memcached.set(listStatus.get(i).statusID, Constants.EXPIRE, listStatus.get(i));
                statusIds.add(listStatus.get(i).statusID);
            }
            //Set list status ID
            Memcached.set("StatusIds", Constants.EXPIRE, statusIds);
            result = true;
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return result;
    }

    public List<String> getListPaging(int limit) {
        List<String> listPaging = new ArrayList<String>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameCreateDate);
            KeyRange range = new KeyRange();
            range.start_key = GeneralHandling.toByteBuffer("");
            range.end_key = GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;

            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            for (int i = 0; i < listKeySlice.size();) {
                listPaging.add(Common.GeneralHandling.toString(listKeySlice.get(i).key));
                i += limit;
            }
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listPaging;
    }

    public List<String> getListPaging(String startKey, int limit) {
        List<String> listPaging = new ArrayList<String>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = new SlicePredicate();
            predicate.addToColumn_names(colNameCreateDate);
            KeyRange range = new KeyRange();
            range.start_key = GeneralHandling.toByteBuffer(startKey);
            range.end_key = GeneralHandling.toByteBuffer("");
            range.count = Short.MAX_VALUE;

            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            for (int i = 0; i < listKeySlice.size();) {
                listPaging.add(Common.GeneralHandling.toString(listKeySlice.get(i).key));
                i += limit;
            }
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listPaging;
    }

    public List<StatusPOJO> selectStatusFromPage(String startKey, int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);

            SlicePredicate predicate = createSlicePredicateForStatus();
            KeyRange range = new KeyRange();
            range.start_key = GeneralHandling.toByteBuffer(startKey);
            range.end_key = GeneralHandling.toByteBuffer("");
            range.count = limit;

            List<KeySlice> listKeySlice = client.get_range_slices(parent, predicate, range, Common.Constants.CL);
            listStatus = getListStatusFromListKeySliceStatus(listKeySlice);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public static List<StatusPOJO> filterNullStatus(List<StatusPOJO> listStatus) {
        int size = listStatus.size();
        for (int i = 0; i < size; i++) {
            if (!listStatus.get(i).isSetContent()) {
                listStatus.remove(i);
                i--;
                size--;
            }
        }
        return listStatus;
    }

    private List<StatusPOJO> SortByLikeAndDislike(List<StatusPOJO> listStatus) throws UnsupportedEncodingException {
        for (int i = 0; i < listStatus.size() - 1; i++) {
            for (int j = i + 1; j < listStatus.size(); j++) {
                float point1 = (float) (listStatus.get(i).likeCount * 0.8 + listStatus.get(i).dislikeCount * 0.2);
                float point2 = (float) (listStatus.get(j).likeCount * 0.8 + listStatus.get(j).dislikeCount * 0.2);
                if (point1 < point2) {
                    StatusPOJO temp = listStatus.get(i);
                    listStatus.set(i, listStatus.get(j));
                    listStatus.set(j, temp);
                }
            }
        }
        return listStatus;
    }

    public static List<StatusPOJO> getListStatusFromListStatusID(List<String> listStatusID, Cassandra.Client client) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();

        ColumnParent parent = new ColumnParent(colParentS);
        SlicePredicate predicate = createSlicePredicateForStatus();
        int sizeListStatus = listStatusID.size();

        for (int i = 0; i < sizeListStatus; i++) {
            String StatusID = listStatusID.get(i);
            StatusPOJO status;
            status = selectStatusWithID(StatusID, client);
            listStatus.add(status);
        }

        listStatus = filterNullStatus(listStatus);

        return listStatus;
    }

    public static List<StatusPOJO> getListStatusFromListStatusIDV1(List<ByteBuffer> listStatusID, Cassandra.Client client) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            ColumnParent parent = new ColumnParent(colParentS);
            SlicePredicate predicate = createSlicePredicateForStatus();

            Map<ByteBuffer, List<ColumnOrSuperColumn>> mapStatus = client.multiget_slice(listStatusID, parent, predicate, Common.Constants.CL);

            for (int i = 0; i < mapStatus.size(); i++) {
                List<ColumnOrSuperColumn> listColOrSupCol = mapStatus.get(listStatusID.get(i));
                String statusID = Common.GeneralHandling.toString(listStatusID.get(i));
                StatusPOJO status = createStatusPOJOFromListColOrSupCol(statusID, listColOrSupCol);
                listStatus.add(status);
            }
            listStatus = filterNullStatus(listStatus);

        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> getListStatusFromListKeySliceStatus(List<KeySlice> listKeySlice) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            for (int i = 0; i < listKeySlice.size(); i++) {
                List<ColumnOrSuperColumn> listColOrSupCol = listKeySlice.get(i).columns;
                String statusID = Common.GeneralHandling.toString(listKeySlice.get(i).key);
                StatusPOJO status = createStatusPOJOFromListColOrSupCol(statusID, listColOrSupCol);
                listStatus.add(status);
            }
            listStatus = filterNullStatus(listStatus);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public List<StatusPOJO> getListStatusFromListKeySliceStatusForRanDom(List<KeySlice> listKeySlice) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {
            for (int i = 0; i < listKeySlice.size(); i++) {
                List<ColumnOrSuperColumn> listColOrSupCol = listKeySlice.get(i).columns;
                String statusID = Common.GeneralHandling.toString(listKeySlice.get(i).key);
                StatusPOJO status = createStatusPOJOFromListColOrSupColForRanDom(statusID, listColOrSupCol);
                listStatus.add(status);
            }
            listStatus = filterNullStatus(listStatus);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    public static SlicePredicate createSlicePredicateForStatus() {
        SlicePredicate predicate = new SlicePredicate();
        predicate.addToColumn_names(colNameContent);
        predicate.addToColumn_names(colNameCreateDate);
        predicate.addToColumn_names(colNameModifyDate);
        predicate.addToColumn_names(colNameLikeCount);
        predicate.addToColumn_names(colNameDislikeCount);
        predicate.addToColumn_names(colNameIsPublic);
        predicate.addToColumn_names(colNameTags);
        return predicate;
    }

    public int getTotal_searchStatusByTag(String keyWord, String filter) throws Exception {
        int count = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status");
        List<StatusPOJO> list = searchStatusByContent(keyWord, "0", Integer.toString(count), filter);
        return list.size();
    }

    public List<StatusPOJO> searchStatusByTag(String value, String start, String end, String filter) throws Exception {
        List<StatusPOJO> result = new ArrayList<StatusPOJO>();
        int _start = Integer.parseInt(start);
        int _end = Integer.parseInt(end);
        int _filter = Integer.parseInt(filter);
        ArrayList<String[]> list = (new solrSearch()).searchStatusByTag(value, _start, _end, _filter);
        for (String[] strings : list) {
            StatusPOJO pojo = createStatusPOJOFromListColOrSupCol(strings[0], GetlistkeysliceFromArray(strings));
            result.add(pojo);
        }
        return result;
    }

    public int getTotal_searchStatusByContent(String keyWord, String filter) throws Exception {
        int count = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status");
        List<StatusPOJO> list = searchStatusByContent(keyWord, "0", Integer.toString(count), filter);
        return list.size();
    }

    public List<StatusPOJO> searchStatusByContent(String value, String start, String end, String filter) throws Exception {
        List<StatusPOJO> result = new ArrayList<StatusPOJO>();
        int _start = Integer.parseInt(start);
        int _end = Integer.parseInt(end);
        int _filter = Integer.parseInt(filter);
        ArrayList<String[]> list = (new solrSearch()).searchStatusByContent(value, _start, _end, _filter);
        for (String[] strings : list) {
            StatusPOJO pojo = createStatusPOJOFromListArray(strings);
            result.add(pojo);
        }
        return result;
    }

    public int getTotal_searchStatusByContentandTag(String keyWord, String tag, String filter) throws Exception {
        int count = (int) AutoIncrementEnvoker.getApi().getNumberIncrement("Status");
        List<StatusPOJO> list = searchStatusByContentandTag(keyWord, tag, "0", Integer.toString(count), filter);
        return list.size();
    }

    public List<StatusPOJO> searchStatusByContentandTag(String value, String Tag, String start, String end, String filter) throws Exception {
        List<StatusPOJO> result = new ArrayList<StatusPOJO>();
        int _start = Integer.parseInt(start);
        int _end = Integer.parseInt(end);
        int _filter = Integer.parseInt(filter);
        ArrayList<String[]> list = (new solrSearch()).searchStatusByContent(value, _start, _end, _filter);
        for (String[] strings : list) {
            if (strings[5].contains(Tag)) {
                StatusPOJO pojo = createStatusPOJOFromListArray(strings);
                result.add(pojo);
            }
        }
        return result;
    }

    private List<ColumnOrSuperColumn> GetlistkeysliceFromArray(String[] list) {
        List<ColumnOrSuperColumn> listkey = new ArrayList<ColumnOrSuperColumn>();


        ColumnOrSuperColumn colsuper0 = new ColumnOrSuperColumn();
        colsuper0.column = Common.GeneralHandling.createCol(colNameContent, list[1]);
        listkey.add(colsuper0);

        ColumnOrSuperColumn colsuper1 = new ColumnOrSuperColumn();
        colsuper1.column = Common.GeneralHandling.createCol(colNameCreateDate, list[2]);
        listkey.add(colsuper1);

        ColumnOrSuperColumn colsuper2 = new ColumnOrSuperColumn();
        colsuper2.column = Common.GeneralHandling.createCol(colNameModifyDate, list[3]);
        listkey.add(colsuper2);

        ColumnOrSuperColumn colsuper3 = new ColumnOrSuperColumn();
        colsuper3.column = Common.GeneralHandling.createCol(colNameIsPublic, list[4]);
        listkey.add(colsuper3);

        ColumnOrSuperColumn colsuper4 = new ColumnOrSuperColumn();
        colsuper4.column = Common.GeneralHandling.createCol(colNameTags, list[5]);
        listkey.add(colsuper4);

        return listkey;
    }

    public static StatusPOJO createStatusPOJOFromListColOrSupCol(String statusID, List<ColumnOrSuperColumn> listColOrSupCol) {
        StatusPOJO status = new StatusPOJO();
        status.statusID = statusID;
        for (int j = 0; j < listColOrSupCol.size(); j++) {
            try {
                ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                Column col = colOrSupCol.column;
                if (col.name.equals(colNameContent)) {
                    status.content = Common.GeneralHandling.toString(col.value);
                }
                if (col.name.equals(colNameCreateDate)) {
                    status.createDate = Common.GeneralHandling.toString(col.value);
                }
                if (col.name.equals(colNameModifyDate)) {
                    status.modifyDate = Common.GeneralHandling.toString(col.value);
                }
                if (col.name.equals(colNameLikeCount)) {
                    status.likeCount = Common.GeneralHandling.toInt(col.value);
                }
                if (col.name.equals(colNameDislikeCount)) {
                    status.dislikeCount = Common.GeneralHandling.toInt(col.value);
                }
                if (col.name.equals(colNameIsPublic)) {
                    status.isPublic = Common.GeneralHandling.toBool(col.value);
                }
                if (col.name.equals(colNameTags)) {
                    status.tags = Common.GeneralHandling.toString(col.value);
                }
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return status;
    }

    public static StatusPOJO createStatusPOJOFromListColOrSupColForRanDom(String statusID, List<ColumnOrSuperColumn> listColOrSupCol) {
        StatusPOJO status = new StatusPOJO();
        status.statusID = statusID;
        for (int j = 0; j < listColOrSupCol.size(); j++) {
            try {
                ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                Column col = colOrSupCol.column;
                if (col.name.equals(colNameContent)) {
                    status.content = Common.GeneralHandling.toString(col.value);
                }
                if (col.name.equals(colNameIsPublic)) {
                    status.isPublic = Common.GeneralHandling.toBool(col.value);
                }
                if (col.name.equals(colNameTags)) {
                    status.tags = Common.GeneralHandling.toString(col.value);
                }
            } catch (UnsupportedEncodingException ex) {
                Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return status;
    }

    public List<String> selectStatusIDFromTag(String tagName) {
        List<String> listStatusID = new ArrayList<String>();
        try {

            ByteBuffer tagNameBB = Common.GeneralHandling.toByteBuffer(tagName);
            ColumnParent parent = new ColumnParent(TagDAO.colParentListStatusS);
            SlicePredicate predicate = new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.start = GeneralHandling.toByteBuffer("");
            sliceRange.finish = GeneralHandling.toByteBuffer("");
            predicate.slice_range = sliceRange;

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(tagNameBB, parent, predicate, Common.Constants.CL);
            for (int j = 0; j < listColOrSupCol.size(); j++) {
                try {
                    ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                    Column col = colOrSupCol.column;
                    if ("1".equals(GeneralHandling.toString(col.value))) {
                        listStatusID.add(GeneralHandling.toString(col.name));
                    }
                } catch (UnsupportedEncodingException ex) {
                    Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
            return listStatusID;

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatusID;
    }

    public List<String> getListPagingOnATag(String tagName, int limit) {
        List<String> listStatusID = new ArrayList<String>();
        try {

            ByteBuffer tagNameBB = Common.GeneralHandling.toByteBuffer(tagName);
            ColumnParent parent = new ColumnParent(TagDAO.colParentListStatusS);
            SlicePredicate predicate = new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.start = GeneralHandling.toByteBuffer("");
            sliceRange.finish = GeneralHandling.toByteBuffer("");
            predicate.slice_range = sliceRange;

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(tagNameBB, parent, predicate, Common.Constants.CL);
            int numberStatus = listColOrSupCol.size();
            for (int j = 0; j < numberStatus;) {
                try {
                    ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                    Column col = colOrSupCol.column;
                    listStatusID.add(GeneralHandling.toString(col.name));
                } catch (UnsupportedEncodingException ex) {
                    Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
                }
                j += limit;
            }
            return listStatusID;

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatusID;
    }

    public List<StatusPOJO> selectStatusFromPageOnATag(String tagName, String startKey, int limit) {
        List<StatusPOJO> listStatus = new ArrayList<StatusPOJO>();
        try {

            ByteBuffer tagNameBB = Common.GeneralHandling.toByteBuffer(tagName);
            ColumnParent parent = new ColumnParent(TagDAO.colParentListStatusS);
            SlicePredicate predicate = new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.start = GeneralHandling.toByteBuffer(startKey);
            sliceRange.finish = GeneralHandling.toByteBuffer("");
            sliceRange.count = limit;
            predicate.slice_range = sliceRange;

            List<ColumnOrSuperColumn> listColOrSupCol = client.get_slice(tagNameBB, parent, predicate, Common.Constants.CL);
            StatusPOJO status = new StatusPOJO();
            for (int j = 0; j < listColOrSupCol.size(); j++) {
                try {
                    ColumnOrSuperColumn colOrSupCol = listColOrSupCol.get(j);
                    Column col = colOrSupCol.column;
                    if ("1".equals(GeneralHandling.toString(col.value))) {
                        status = selectStatusWithID(GeneralHandling.toString(col.name));
                        listStatus.add(status);
                    }
                } catch (UnsupportedEncodingException ex) {
                    Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
                }
            }

        } catch (InvalidRequestException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            Logger.getLogger(StatusDAO.class.getName()).log(Level.SEVERE, null, ex);
        }
        return listStatus;
    }

    private StatusPOJO createStatusPOJOFromListArray(String[] strings) {
        StatusPOJO status = new StatusPOJO();
        status.statusID = strings[0];
        status.content = strings[1];
        status.createDate = strings[2];
        status.modifyDate = strings[3];
        status.isPublic = ("1".equals(strings[4])) ? true : false;
        status.tags = strings[5];
        return status;
    }
}